package com.ccnu.demojava.flink;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.time.Duration;

/**
 * 一般开发流程
 * 1.生成一个datastream或者KeyedStream
 * 2.定义一组规则
 * 3.将这组规则应用于流，生成PatternStream
 * 4.将生成的PatternStream，通过select方法，将符合规则的数据通过自定义的输出形式，生成结果流
 * 5.结果流就是我们最终得到的数据
 */
public class FlinkCepTest {
    private static final Logger LOGGER = (Logger) LoggerFactory.getLogger(FlinkCepTest.class);

    public static Pattern<LoginEvent, LoginEvent> FailTreePattern() {
        // 在3秒 内重复登录了三次, 则产生告警
        Pattern<LoginEvent, LoginEvent> loginFailPattern = Pattern.<LoginEvent>
                        begin("first")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context context) throws Exception {
                        System.out.println("first: " + loginEvent.toString());
                        return loginEvent.getType().equals("fail");
                    }
                })
                .next("second")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context context) throws Exception {
                        System.out.println("second: " + loginEvent);
                        return loginEvent.getType().equals("fail");
                    }
                })
                .next("three")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context context) throws Exception {
                        System.out.println("three: " + loginEvent);
                        return loginEvent.getType().equals("fail");
                    }
                })
                .within(Duration.ofSeconds(3));

        return loginFailPattern;
    }

    public static Pattern<LoginEvent, LoginEvent> SimplePattern() {
        return Pattern.<LoginEvent>begin("s1")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        System.out.println("filter_value:" + value.toString());
                        return true;
                    }
                });
    }

    public static void main(String[] args) throws Exception {
        // 环境 + 数据源
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<LoginEvent> loginEventStream = env.fromData(Arrays.asList(
                new LoginEvent("小明", "192.168.0.1", "fail"),
                new LoginEvent("小明", "192.168.0.2", "fail"),
                new LoginEvent("小王", "192.168.10,11", "fail"),
                new LoginEvent("小王", "192.168.10,12", "fail"),
                new LoginEvent("小明", "192.168.0.3", "fail"),
                new LoginEvent("小明", "192.168.0.4", "fail"),
                new LoginEvent("小王", "192.168.10,10", "success")
        ));

        // 定义模式
        Pattern<LoginEvent, LoginEvent> patternTemp = SimplePattern();

        // 根据用户id分组，以便可以锁定用户IP，cep模式匹配
        PatternStream<LoginEvent> patternStream = CEP.pattern(
                loginEventStream.keyBy(LoginEvent::getUserId),
                patternTemp);

//        DataStream<String> tDataStream = patternStream.select(
//                new PatternSelectFunction<LoginEvent, String>() {
//                    @Override
//                    public String select(Map<String, List<LoginEvent>> pattern) throws Exception {
//                        System.out.println("result size:" + new String(String.valueOf(pattern.size())));
//                        List<LoginEvent> second = pattern.get("s1");
//                        System.out.println("result:" + second.get(0).toString());
//                        return "result:" + second.get(0).getUserId() + ", " + second.get(0).getIp() + ", " + second.get(0).getType();
//                    }
//                }
//        );

        DataStream<String> tDataStream = patternStream.process(
                new PatternProcessFunction<LoginEvent, String>() {
                    @Override
                    public void processMatch(Map<String, List<LoginEvent>> map, Context context, Collector<String> collector) throws Exception {
                        System.out.println("proces:" + String.valueOf(map.size()));
//                        collector.collect();
                    }
                }
        );

        tDataStream.sinkTo(new Sink<String>() {
            @Override
            public SinkWriter<String> createWriter(InitContext context) throws IOException {
                return null;
            }

            @Override
            public SinkWriter<String> createWriter(WriterInitContext context) throws IOException {
                return Sink.super.createWriter(context);
            }
        });

        // 打印告警用户
        tDataStream.print();
        env.execute("flink cep test");
        LOGGER.info("finish");
    }
}
